package ru.spbau.database.group3.index;

import ru.spbau.database.group3.engine.*;
import ru.spbau.database.group3.filesAndAccess.Page;
import ru.spbau.database.group3.filesAndAccess.Record;
import ru.spbau.database.group3.memoryManager.SpaceManager;

import java.io.IOException;
import java.io.Serializable;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * User: stels
 * Date: 08.02.13
 * Time: 23:13
 */
public class HashIndex implements Serializable {

    public HashIndex(String path, int recordSize, List<Column> columns) throws IOException {
        this.path = path;
        this.recordSize = recordSize;
        this.globalDepth = 1;
        this.columns = columns;
        this.directory = new ArrayList<Bucket>();
        this.directory.add(allocFirstBucket());
    }

    public List<Record> select(List<Condition> searchKey, List<Condition> conditions) throws IOException {
        int index = defineBucket(searchKey);
        List<Record> indexEntries = directory.get(index).getContent().getRecords(searchKey);
        List<Record> result = new ArrayList<Record>();
        for(Record r : indexEntries) {
            Record dataRecord = getDataRecord(r);
            if(dataRecord.satisfy(conditions)) {
                result.add(dataRecord);
            }
        }
        return result;
    }

    public List<Record> delete(List<Condition> searchKey, List<Condition> conditions) throws IOException {
        int index = defineBucket(searchKey);
        return directory.get(index).getContent().deleteRecords(conditions);
    }

    public void insert(ByteBuffer entryData) throws IOException {
        List<Object> key = getKey(entryData);
        int index = defineBucket(hash(key));
        if(directory.get(index).isFull()) {
            if(directory.get(index).getLocalDepth() == globalDepth) {
                doubleDirectory();
            }
            splitBucket(index);
            insert(entryData);
        }
        directory.get(index).insertRecord(entryData);
    }


    public List<Object> getKey(Record entry) throws IOException {
        return entry.getColumnsValue(columns.subList(0, columns.size() - 2));
    }

    public String getPath() {
        return path;
    }

    public List<Object> getKey(ByteBuffer entry) throws IOException {
        ArrayList<Object> key = new ArrayList<Object>();
        for(Column c : columns.subList(0, columns.size() - 2)) {
            key.add(c.getType().readValueFromByteBuffer(entry, c.getOffset()));
        }
        return key;
    }

    public static int remainder(int dividend, int divisor) {
        int rem = dividend - divisor * (dividend / divisor);
        return rem < 0 ? -rem : rem;
    }

    private void doubleDirectory() {
        globalDepth *= 2;
        directory.addAll(directory);
    }


    private int defineBucket(int hashCode) {
        return remainder(hashCode, globalDepth);
    }

    private int defineBucket(List<Condition> searchKey) {
        List<Object> keyValue = new ArrayList<Object>();
        for(Condition c : searchKey) {
            keyValue.add(c.getValue());
        }
        return defineBucket(hash(keyValue));
    }

    private int hash(List<Object> value) {
        return 31 * value.hashCode() + 29;
    }

    private ByteBuffer getEntryData(Record entry) throws IOException {
        return ByteBuffer.wrap(entry.toByteArray());
    }

    private void distributeRecords(Bucket src, Bucket split) throws IOException {
        for(Record entry : src.getContent().getAllRecords()) {
            int index = defineBucket(hash(getKey(entry)));
            if(index != src.getId()) {
                split.insertRecord(getEntryData(entry));
                src.deleteRecord(entry);
            }
        }
    }

    private Bucket allocFirstBucket() throws IOException {
        Page firstPage = SpaceManager.allocPage(path + Relation.DATA_EXT, recordSize);
        return new Bucket(firstPage, 1, 0);
    }

    private Bucket getSplitBucket(Bucket src) throws IOException {
        int splitId = src.getId() + src.getLocalDepth();
        Page splitPage = SpaceManager.allocPage(path + Relation.DATA_EXT, recordSize);
        return new Bucket(splitPage, src.getLocalDepth() * 2, splitId);
    }



    private void splitBucket(int index) throws IOException {
        Bucket src = directory.get(index);
        Bucket split = getSplitBucket(src);
        src.doubleDepth();
        distributeRecords(src, split);
        directory.set(split.getId(), split);
    }

    private Record getDataRecord(Record indexEntry) throws IOException {
        int pid = (Integer) indexEntry.getColumnValue(columns.get(columns.size() - 2));
        int sid = (Integer) indexEntry.getColumnValue(columns.get(columns.size() - 1));
        return new Record(pid, sid);
    }

    private final String path;
    private int recordSize;
    private int globalDepth;
    private List<Column> columns;
    private List<Bucket> directory;





}